package top.luoyuanxiang.mqtt.config;

import jakarta.annotation.Resource;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.common.MqttSubscription;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.mqtt.inbound.Mqttv5PahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.Mqttv5PahoMessageHandler;
import org.springframework.integration.mqtt.support.MqttHeaderMapper;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;

import java.util.List;
import java.util.UUID;

/**
 * mqtt 配置(消费端、服务端)
 *
 * @author luoyuanxiang
 */
@Configuration
public class MqttConfig {

    @Resource
    private MqttProperties mqttProperties;

    /**
     * MQTT 连接选项
     *
     * @return {@link MqttConnectionOptions }
     */
    @Bean
    public MqttConnectionOptions mqttConnectionOptions() {
        MqttConnectionOptions connectionOptions = new MqttConnectionOptions();
        connectionOptions.setUserName(mqttProperties.getUsername());
        connectionOptions.setPassword(mqttProperties.getPassword().getBytes());
        connectionOptions.setServerURIs(new String[]{mqttProperties.getBroker()});
        connectionOptions.setKeepAliveInterval(mqttProperties.getKeepAliveInterval());
        // 断线重连，默认没有重新订阅，必须重写 connectComplete 方法
        connectionOptions.setAutomaticReconnect(mqttProperties.isAutomaticReconnect());
        // 是否重新创建 session
        connectionOptions.setCleanStart(mqttProperties.isCleanStart());
        connectionOptions.setConnectionTimeout(mqttProperties.getConnectionTimeout());
        connectionOptions.setSessionExpiryInterval(mqttProperties.getSessionExpiryInterval());
        return connectionOptions;
    }

    /**
     * MQTT 5 客户端消息监听配置
     *
     * @return {@link MessageChannel }
     */
    @Bean
    public MessageChannel mqtt5InboundChannel() {
        return MessageChannels.publishSubscribe().getObject();
    }

    /**
     * MQTT 5 服务端消息监听配置
     *
     * @return {@link MessageChannel }
     */
    @Bean
    public MessageChannel mqtt5OutboundChannel() {
        return MessageChannels.direct().getObject();
    }

    /**
     * MQTT 5 服务端配置
     *
     * @return {@link MessageHandler }
     */
    @Bean
    @ServiceActivator(inputChannel = "mqtt5OutboundChannel")
    public MessageHandler mqtt5Outbound() {
        Mqttv5PahoMessageHandler messageHandler = new Mqttv5PahoMessageHandler(mqttConnectionOptions(), "produce_" + UUID.randomUUID().toString().replace("-", ""));
        MqttHeaderMapper mqttHeaderMapper = new MqttHeaderMapper();
        mqttHeaderMapper.setOutboundHeaderNames(MqttHeaders.RESPONSE_TOPIC, MqttHeaders.CORRELATION_DATA, MessageHeaders.CONTENT_TYPE);
        messageHandler.setHeaderMapper(mqttHeaderMapper);
        messageHandler.setDefaultTopic("test");
        messageHandler.setDefaultQos(1);
        return messageHandler;
    }

    /**
     * MQTT V5 消费端配置
     *
     * @return {@link MessageProducer }
     */
    @Bean
    public MessageProducer mqtt5Inbound() {
        List<MqttProperties.Topic> topicList = mqttProperties.getTopicList();
        List<MqttSubscription> mqttSubscriptionList = topicList.stream().map(s -> new MqttSubscription(s.getName(), s.getQos())).toList();
        MqttSubscription[] mqttSubscriptions = mqttSubscriptionList.toArray(MqttSubscription[]::new);
        Mqttv5PahoMessageDrivenChannelAdapter adapter = new Mqttv5PahoMessageDrivenChannelAdapter(mqttConnectionOptions(), "consumer_" + UUID.randomUUID().toString().replace("-", ""), mqttSubscriptions);
        adapter.setOutputChannel(mqtt5InboundChannel());
        return adapter;
    }
}
